/*
 * JuiceFS, Copyright (C) 2020 Juicedata, Inc.
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

package redis

import (
	"context"
	"fmt"
	"net"
	"os"
	"sort"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/go-redis/redis/v8"
)

const aclbaseString = "ACL SETUSER ${USERNAME} on >${PASSWORD} ~${USERNAME}:* +@all -@admin -@dangerous"

// RedisConfig is config for Redis client.
type RedisConfig struct {
	Strict  bool // update ctime
	Retries int
}

type redisMeta struct {
	sync.Mutex
	conf    *RedisConfig
	rdb     *redis.Client
	txlocks [1024]sync.Mutex // Pessimistic locks to reduce conflict on Redis

	sid          int64
	openFiles    map[Ino]int
	removedFiles map[Ino]bool
	compacting   map[uint64]bool
	symlinks     *sync.Map
	msgCallbacks *msgCallbacks

	shaLookup string // The SHA returned by Redis for the loaded `scriptLookup`
}

var _ Meta = &redisMeta{}

type msgCallbacks struct {
	sync.Mutex
	callbacks map[uint32]MsgCallback
}

//func genKey(key string) string {
//	return os.Getenv("JUICEFS_NAME") + ":" + key
//}

// NewRedisMeta return a meta store using Redis.
func NewRedisMeta(url string, conf *RedisConfig) (Meta, error) {
	opt, err := redis.ParseURL(url)
	if err != nil {
		return nil, fmt.Errorf("parse %s: %s", url, err)
	}
	var rdb *redis.Client
	fmt.Println(opt.Addr)
	if strings.Contains(opt.Addr, ",") {
		var fopt redis.FailoverOptions
		h, _, _ := net.SplitHostPort(opt.Addr)
		opt.Addr = h
		ps := strings.Split(opt.Addr, ",")
		fopt.MasterName = ps[0]
		fopt.SentinelAddrs = ps[1:]
		_, port, _ := net.SplitHostPort(fopt.SentinelAddrs[len(fopt.SentinelAddrs)-1])
		if port != "" {
			for i := range fopt.SentinelAddrs {
				h, p, _ := net.SplitHostPort(fopt.SentinelAddrs[i])
				if p == "" {
					fopt.SentinelAddrs[i] = net.JoinHostPort(h, port)
				}
			}
		}
		// Assume Redis server and sentinel have the same password.
		fopt.SentinelPassword = os.Getenv("SENTINEL_PASSWORD")
		fopt.Username = opt.Username
		fopt.Password = opt.Password

		fmt.Println(fopt.SentinelPassword)
		fmt.Println(fopt.Password)

		if fopt.SentinelPassword == "" && os.Getenv("SENTINEL_PASSWORD") != "" {
			fopt.SentinelPassword = os.Getenv("SENTINEL_PASSWORD")
		}
		if fopt.Password == "" && os.Getenv("REDIS_PASSWORD") != "" {
			fopt.Password = os.Getenv("REDIS_PASSWORD")
		}
		fopt.DB = opt.DB
		fopt.TLSConfig = opt.TLSConfig
		fopt.MaxRetries = conf.Retries
		fopt.MinRetryBackoff = time.Millisecond * 100
		fopt.MaxRetryBackoff = time.Minute * 1
		fopt.ReadTimeout = time.Second * 30
		fopt.WriteTimeout = time.Second * 5
		rdb = redis.NewFailoverClient(&fopt)
	} else {
		if opt.Password == "" && os.Getenv("REDIS_PASSWORD") != "" {
			opt.Password = os.Getenv("REDIS_PASSWORD")
		}
		opt.MaxRetries = conf.Retries
		opt.MinRetryBackoff = time.Millisecond * 100
		opt.MaxRetryBackoff = time.Minute * 1
		opt.ReadTimeout = time.Second * 30
		opt.WriteTimeout = time.Second * 5
		rdb = redis.NewClient(opt)
	}
	m := &redisMeta{
		conf:         conf,
		rdb:          rdb,
		openFiles:    make(map[Ino]int),
		removedFiles: make(map[Ino]bool),
		compacting:   make(map[uint64]bool),
		symlinks:     &sync.Map{},
		msgCallbacks: &msgCallbacks{
			callbacks: make(map[uint32]MsgCallback),
		},
	}

	return m, nil
}

func errno(err error) syscall.Errno {
	if err == nil {
		return 0
	}
	if eno, ok := err.(syscall.Errno); ok {
		return eno
	}
	if err == redis.Nil {
		return syscall.ENOENT
	}
	//logger.Errorf("error: %s", err)
	return syscall.EIO
}

func (r *redisMeta) LoadACL(ctx context.Context) error{
	_, err := r.rdb.Do(ctx, "ACL", "LOAD").Result()
	if err != nil {
		return err
	}
	return nil
}

func (r *redisMeta) SaveACL(ctx context.Context) error {
	_, err := r.rdb.Do(ctx, "ACL", "SAVE").Result()
	if err != nil {
		return err
	}
	return nil
}

func (r *redisMeta) GetACLUserList(ctx context.Context) ([]string, error) {
	vals, err := r.rdb.Do(ctx, "ACL", "USERS").Result()
	if err != nil {
		return nil, err
	}
	userValue, _ := vals.([]interface{})

	users := make([]string, len(userValue))
	for i, arg := range userValue {
		users[i] = arg.(string)
	}
	//fmt.Println(users)
	return users, nil
}

func (r *redisMeta) SetACLUser(ctx context.Context, username, password string) error {
	//aclbaseString := "ACL SETUSER test2 on >test2 ~test2:* +@all -@admin -@dangerous"

	// Step1. LOAD
	err := r.LoadACL(ctx)

	if err != nil {
		return err
	}

	// Step2. Get
	users, _ := r.GetACLUserList(ctx)

	//fmt.Println(users)

	if in(username, users){
		return fmt.Errorf("user is exists")
	}

	// Step3. Set
	aclUserString := strings.Replace(aclbaseString, "${USERNAME}", username, 2)
	aclUserString = strings.Replace(aclUserString, "${PASSWORD}", password, 1)

	cmdArgs := strings.Split(aclUserString, " ")

	s := make([]interface{}, len(cmdArgs))
	for i, v := range cmdArgs {
		s[i] = v
	}

	_, err = r.rdb.Do(ctx, s...).Result()
	if err != nil {
		return err
	}

	// Step4. Save
	err = r.SaveACL(ctx)

	if err != nil {
		return err
	}

	return nil
}

func in(target string, str_array []string) bool {
	sort.Strings(str_array)
	index := sort.SearchStrings(str_array, target)
	//index的取值：[0,len(str_array)]
	if index < len(str_array) && str_array[index] == target {
		//需要注意此处的判断，先判断 &&左侧的条件，如果不满足则结束此处判断，不会再进行右侧的判断
		return true
	}
	return false
}
